package me.tuyou.wordcount;

import me.tuyou.common.WordStreamGenerator;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

public class TableStreamWordCountTest {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useOldPlanner()
                .inStreamingMode()
                .build();

        StreamTableEnvironment stEnv = StreamTableEnvironment.create(env, settings);
        DataStream<String> dataStream = env.addSource(new WordStreamGenerator());
        Table table = stEnv.fromDataStream(dataStream, $("word"))
                .where($("word").like("%t%"));
        String old = stEnv.explain(table);
        System.out.println(old);

        stEnv.toAppendStream(table, Row.class).print("table");
        env.execute();
    }
}
